package com.colorlight.cloud.gateway.core.message.instruct;

import com.colorlight.cloud.common.exception.BaseException;
import com.colorlight.cloud.common.exception.ParameterException;
import com.colorlight.cloud.gateway.api.message.constraint.schedule.ScheduleActions;
import com.colorlight.cloud.gateway.api.message.convert.CommandConvert;
import com.colorlight.cloud.gateway.api.message.instruct.api.MessageLiteInfo;
import com.colorlight.cloud.gateway.api.message.instruct.api.PublishCommand;
import com.colorlight.cloud.gateway.api.message.model.CommandData;
import com.colorlight.cloud.mqtt.base.api.message.PublishMessage;
import com.colorlight.cloud.mqtt.base.entity.IMqttHeaders;
import com.google.protobuf.MessageLite;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.*;

import java.util.*;

import static com.colorlight.cloud.gateway.core.model.GatewayCoreExceptionEnum.*;


/**
 * @author Demon
 * 消息下发的具体实现
 */
@Component
@Slf4j
public class PublishCommandImpl implements PublishCommand {
	
	private final MessageLiteInfo messageLiteInfo;
	
	private final PublishMessage publishMessage;
	
	public PublishCommandImpl(MessageLiteInfo messageLiteInfo, PublishMessage publishMessage) {
		this.messageLiteInfo = messageLiteInfo;
		this.publishMessage = publishMessage;
	}
	
	
	@Override
	public Long publishCommandData(CommandData<?> commandData) {
		
		//默认同指令得型号相同
		Long seqId = commandData.getSeqId();
		MessageLite convert = null;
		if (Objects.isNull(seqId)) {
			convert = CommandConvert.convert(commandData.getContent(), commandData.getUuid());
		} else {
			convert = CommandConvert.convert(commandData.getContent(), commandData.getUuid(), seqId);
		}
		seqId = this.publishMessageLite(Collections.singleton(commandData.getTopic()), convert);
		commandData.setSeqId(seqId);
		log.info("指令下发成功:{}", commandData);
		return seqId;
	}
	
	
	@Override
	public Long publishMessageLite(Collection<String> topics, MessageLite msg) {
		if (CollectionUtils.isEmpty(topics) || ObjectUtils.isEmpty(msg)) {
			log.error("topic or msg is incorrectness！");
			return null;
		}
		for (String topic : topics) {
			publishCommand(topic, msg);
		}
		return messageLiteInfo.getMessageLiteSeqId(msg);
	}
	
	
	@Override
	public MessageLite packInstruction(ScheduleActions scheduleActions, String uuid, Long seqId) {
		//默认同指令得型号相同
		return CommandConvert.convert(scheduleActions, uuid, seqId);
	}
	
	/**
	 * 消息下发执行者
	 *
	 * @param topic 主题
	 * @param msg   转为protobuf对象的消息
	 **/
	private void publishCommand(String topic, MessageLite msg) {
		
		if (!StringUtils.hasText(topic)) {
			throw new ParameterException(TOPIC_INVALID_PARAMETER_EXCEPTION);
		}
		if (ObjectUtils.isEmpty(msg)) {
			throw new ParameterException(MESSAGE_INVALID_PARAMETER_EXCEPTION);
		}
		//序列化之后下发
		byte[] bytes = msg.toByteArray();
		IMqttHeaders<byte[]> iMqttHeaders = new IMqttHeaders<>(topic, bytes);
		try {
			publishMessage.publish(iMqttHeaders);
		} catch (Exception e) {
			throw new BaseException(INSTRUCTION_DELIVERY_EXCEPTION);
		}
	}
	
}
